MacOS 单机搭建 Pulsar 集群
作者：高天赐
编辑：Irene
Pulsar 是一个支持多租户的、高性能的消息中间件，本文详细记录了如何在 MacOS 上，利用一台服务器搭建一个简单测试集群的过程。后续我们会介绍如何在多台 Linux 服务器搭建集群。
一、准备资源
一台主机（本文以 MacOS 为例）
JDK 8 运行环境（默认已准备好）
二、集群的组成
ZooKeeper 集群（3 个 ZooKeeper 节点组成）
bookie 集群（3 个 BookKeeper 节点组成）
broker 集群（3 个 Pulsar 节点组成）
三、搭建 ZooKeeper 集群
本小节记录了如何在一台机器上运行多个 ZooKeeper 服务进程，搭建 ZooKeeper 集群的过程。
3.1 下载 ZooKeeper 安装包
从 ZooKeeper 官网下载最新版本的 ZooKeeper。本文以 ZooKeeper 3.4.14 版本为例。
3.2 创建 ZooKeeper 集群
创建 zookeepers 文件夹，把解压好的 zookeeper-3.4.14 文件夹复制到 zookeepers 文件夹中，重名为 server1。用 server1 复制出 server2 和 server3。
3.3 配置 ZooKeeper 集群一个节点：
server1
1. 在 server1 目录下，创建 data 和 dataLog 两个文件夹。
2. 在 server1/conf 目录中，复制 zoo_sample.cfg 文件，重命名为 zoo.cfg 文件。
3. 修改 zoo.cfg 文件中的 dataDir, dataLogDir, clientPort, admin.enableServer 和 admin.serverPort 这 5 个配置参数，并添加如下集群节点信息。修改端口号是为了避免在一台服务器上造成端口号冲突。
# 添加如下集群节点信息server.1=127.0.0.1:2888:3888
server.2=127.0.0.1:2889:3889
server.3=127.0.0.1:2890:3890
zoo.cfg 文件配置说明
在一台服务器上，部署多个实例，需要指定不同的端口号。
clientPort: 3 个 ZooKeeper 节点中分别配置为:2181，2182，2183。
admin.enableServer: 3 个 ZooKeeper 节点中都配置为 true。
admin.serverPort: 3 个 ZooKeeper 节点中分别配置为:9181，9182，9183。
集群节点信息：3 个 ZooKeeper 节点中都配置为:
server.1=127.0.0.1:2888:3888
server.2=127.0.0.1:2889:3889
server.3=127.0.0.1:2890:3890
oo.cfg 配置文件示例
如下是一个 zoo.cfg 文件示例。
# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between
# sending a request and getting an acknowledgement
syncLimit=10
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just
# example sakes.
dataDir=/Users/bilahepan/Softwares/zookeepers/server1/data
dataLogDir=/Users/bilahepan/Softwares/zookeepers/server1/dataLog
# the port at which the clients will connect
clientPort=2181
# the maximum number of client connections.
# increase this if you need to handle more clients
#maxClientCnxns=60
admin.enableServer=true
admin.serverPort=9181
#
server.1=127.0.0.1:2888:3888
server.2=127.0.0.1:2889:3889
server.3=127.0.0.1:2890:3890
# Be sure to read the maintenance section of the
# administrator guide before turning on autopurge.
#
#http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
#
# The number of snapshots to retain in dataDir
#autopurge.snapRetainCount=3
# Purge task interval in hours
# Set to "0" to disable auto purge feature
#autopurge.purgeInterval=1
zoo.cfg 配置参数解释
tickTime：ZooKeeper 中使用的基本时间单位, 毫秒值，默认是 2000ms。
initLimit：用来配置 ZooKeeper 服务器集群中 Follower 服务器初始化连接 Leader 服务器时最长能忍受多少个 tickTime。这里设置为 10 表示最长容忍时间为 10s。
syncLimit：用来配置 Leader 与 Follower 之间发送消息、请求和应答时间最长能忍受多少个 tickTime。这里设置为 10 表示最长容忍时间为 10s。
dataDir：数据文件目录。
dataLogDir：日志文件目录。
clientPort：监听client连接的端口号。
server.{myid}={ip}:{leader服务器交换信息的端口}:{当leader服务器挂了后, 选举leader的端口}
maxClientCnxns：对于一个客户端的连接数限制，默认是 60。
admin.enableServer：是否启用 ZooKeeper 管理后台。
admin.serverPort：管理后台端口号。
4. 在 server1/data/ 目录下创建名字为 myid 的文件，并写入内容 1。每个 ZooKeeper 节点的 myid 文件内容不能一样，它是不同节点的唯一标识。3 个 ZooKeeper 节点中 myid 文件内容分别为1，2，3。
3.4 配置 ZooKeeper 另外两个节点：
server2，server3
对 ZooKeeper 的另外两个节点 server2，server3 做相应配置。具体操作，参考 server1 配置。
启动 ZooKeeper
启动 ZooKeeper 集群，在终端分别用命令启动 ZooKeeper 节点。
# 启动命令(zkServer.sh start)
./bin/zkServer.sh start
查看节点状态。
# 状态查看命令(zkServer.sh status)
./bin/zkServer.sh status
ZooKeeper 的其他操作命令。
# 连接 zookeeper 客户端
./bin/zkCli.sh -timeout 5000 -server 127.0.0.1:2181
# 查看 zookeeper 节点内容
ls /
四、搭建 bookie 集群
Bookie 是 BookKeeper 的别称。本小节记录了如何在一台机器上运行多个 bookie 服务进程，搭建 bookie 集群的过程。
4.1 下载 BookKeeper 安装包
从 BookKeeper 官网下载 BookKeeper 安装包。本文以 BookKeeper 4.8.2 版本为例。
4.2 创建 BookKeeper 集群
创建 bookkeepers 文件夹，把解压好的 bookkeeper-server-4.8.2 文件夹复制到 bookkeepers 文件夹中，重名为 bookie1。用 bookie1 复制出bookie2 和 bookie3。
4.3 配置 BookKeeper 集群一个节点：
bookie1
在 bookie1/conf 目录下，修改 bk_server.conf 文件内容，主要修改 bookiePort，httpServerPort，storageserver.grpc.port 这 3 个端口号，避免在一台主机上造成端口号冲突。添加 ZooKeeper 集群节点信息，并指定两个文件目录地址。
# 添加如下ZooKeeper 集群节点信息zkServers=localhost:2181,localhost:2182,localhost:2183
# 指定两个文件目录地址journalDirectories=/Users/<username>/Softwares/bookkeepers/bookie1/tmp/bk-txn
ledgerDirectories=/Users/<username>/Softwares/bookkeepers/bookie1/tmp/bk-data
bk_server.conf 文件配置说明
在一台服务器上，部署多个实例，需要指定不同的端口号。
bookiePort: 3个 bookie 节点中分别配置为:3181，3182，3183。
httpServerPort: 3个 bookie 节点中都配置为:8050，8060，8070。
storageserver.grpc.port: 3个 bookie 节点中都配置为: 4181，4182，4183。
zk集群节点信息: 3个 bookie 中都配置为: zkServers=localhost:2181,localhost:2182,localhost:2183
4.4 配置 BookKeeper 另外两个节点：
bookie2，bookie3
对 BookKeeper 另外两个节点 bookie2，bookie3 做对应配置。具体操作参考 bookie1 配置。
4.5 初始化集群
在任一个 bookie 中运行以下命令，初始化集群元数据。
./bin/bookkeeper shell metaformat
4.6 启动 BookKeeper 集群
在终端分别用如下命令启动三个 bookie 节点，启动 BookKeeper 集群。
./bin/bookkeeper bookie
4.7 检查 BookKeeper 启动状态
在任一个 bookie 实例中运行以下命令，检查 BookKeeper 启动状态。
集群启动成功会有 "Bookie sanity test succeeded" 日志输出。
./bin/bookkeeper shell bookiesanity
五、搭建 broker 集群
broker 是 Pulsar 实例别称。本小节记录了如何在一台机器上运行多个 broker 服务进程，搭建 broker 集群的过程。
5.1 下载 Pulsar 安装包
从 Pulsar官网下载最新版本的 Pulsar 安装包。本文以 Pulsar 2.3.1 版本为例。
5.2 创建 broker 集群
创建 brokers 文件夹，把解压好的 apache-pulsar-2.3.1 复制到 brokers 文件夹中，重名为 broker1。用 broker1 复制出 broker2 和 broker3
5.3 配置 broker 集群一个节点：
broker1
在 broker1/conf 目录中，修改 broker.conf 文件内容，主要修改 brokerServicePort，brokerServicePortTls，webServicePort，webServicePortTls 这 4 个端口号，避免在一台主机上造成端口号冲突。添加 ZooKeeper 集群节点信息，指定集群名。
# 添加如下 ZooKeeper 集群节点信息
zookeeperServers=localhost:2181,localhost:2182,localhost:2183
configurationStoreServers=localhost:2181,localhost:2182,localhost:2183
# 指定集群名
clusterName=pulsar-cluster
broker.conf 文件配置说明
在一台服务器上，部署多个实例，需要指定不同的端口号。
brokerServicePort: 3 个 broker 节点中分别配置为:6650，6660，6670。
brokerServicePortTls: 3 个 broker 节点中分别配置为:6651，6661，6671。
webServicePort: 3 个 broker 节点中分别配置为:8080，8081，8082。
webServicePortTls: 3 个 broker 节点中分别配置为:8443，8444，8445。
zk集群节点信息: 3 个 broker 中都配置为:
zookeeperServers=localhost:2181,localhost:2182,localhost:2183 configurationStoreServers=localhost:2181,localhost:2182,localhost:2183
5.4 配置 broker 集群另外两个节点：
broker2，broker3
对 broker 集群另外两个节点 broker2，broker3 做相应配置。具体操作，参考 broker1 配置。
5.5 初始化集群
在 broker1 中运行以下命令，初始化集群元数据。
bin/pulsar initialize-cluster-metadata \
--cluster pulsar-cluster \
--zookeeper 127.0.0.1:2181 \
--configuration-store 127.0.0.1:2181 \
--web-service-url http://pulsar.cluster.com:8080 \
--web-service-url-tls https://pulsar.cluster.com:8443 \
--broker-service-url pulsar://pulsar.cluster.com:6650 \
--broker-service-url-tls pulsar+ssl://pulsar.cluster.com:6651
集群元数据解释
--cluster
集群名称
--zookeeper
ZooKeeper集群连接参数，仅需要包含集群中的一个节点即可
--configuration-store
Pulsar实例的配置存储集群（ZooKeeper），和--zookeeper参数一样只需要包含集群中的一个节点即可
--web-service-url
集群Web服务的URL+端口，URL必须是一个标准的DNS名称，默认端口8080，不建议修改。
--web-service-url-tls
集群Web提供TLS服务的URL+端口，端口默认8443，不建议修改。
--broker-service-url
集群brokers服务URL，URL中DNS的名称和Web服务保持一致，URL使用pulsar替代http/http，端口默认6650，不建议修改。
--broker-service-url-tls
集群brokers提供TLS服务的URL，默认端口6551，不建议修改。
5.6 启动 broker 集群
在终端分别用如下命令启动三个 broker 节点，启动 broker 集群。启动成功后会有日志 “PulsarService started” 输出。
./bin/pulsar broker
六、测试 demo
成功启动 Pulsar 集群后，用如下命令依次创建集群、租户、命名空间、分区topic，并为命名空间指定集群。
# 创建集群(集群名:pulsar-cluster)
./bin/pulsar-admin clusters create --url http://pulsar.cluster.com:8080 pulsar-cluster
# 创建租户(租户名:my-tenant)
./bin/pulsar-admin tenants create my-tenant
# 创建命名空间(命名空间名,指定了租户my-tenant:my-tenant/my-namespace)
./bin/pulsar-admin namespaces create my-tenant/my-namespace
# 创建持久性分区topic(topic全名:persistent://my-tenant/my-namespace/my-topic)
./bin/pulsar-admin topics create-partitioned-topic persistent://my-tenant/my-namespace/my-topic -p 3
# 更新命名空间为其指定集群
./bin/pulsar-admin namespaces set-clusters my-tenant/my-namespace --clusters pulsar-cluster
设置 maven 依赖
<dependency>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-client</artifactId>
<!-- 指定你的版本 -->
<version>2.2.0</version>
</dependency>
创建生产者
public class PulsarProducerDemo { private static String localClusterUrl = "pulsar://localhost:6650"; public static void main(String[] args) { try { Producer<byte[]> producer = getProducer(); String msg = "hello world pulsar!"; Long start = System.currentTimeMillis(); MessageId msgId = producer.send(msg.getBytes()); System.out.println("spend=" + (System.currentTimeMillis() - start) + ";send a message msgId = " + msgId.toString());
} catch (Exception e) { System.err.println(e);
}
} public static Producer<byte[]> getProducer() throws Exception { PulsarClient client;
client = PulsarClient.builder().serviceUrl(localClusterUrl).build(); Producer<byte[]> producer = client.newProducer().topic("persistent://my-tenant/my-namespace/my-topic").producerName("producerName").create(); return producer;
}
}
创建消费者
public class PulsarConsumerDemo { private static String localClusterUrl = "pulsar://localhost:6650"; public static void main(String[] args) { try { //将订阅消费者指定的主题和订阅
Consumer<byte[]> consumer = getClient().newConsumer()
.topic("persistent://my-tenant/my-namespace/my-topic")
.subscriptionName("my-subscription")
.subscribe(); while (true) { Message msg = consumer.receive(); System.out.printf("consumer-Message received: %s. \n", new String(msg.getData())); // 确认消息，以便broker删除消息
consumer.acknowledge(msg);
}
} catch (Exception e) { System.out.println(e);
}
} public static PulsarClient getClient() throws Exception { PulsarClient client;
client = PulsarClient.builder().serviceUrl(localClusterUrl).build(); return client;
}
}
更多关于 Pulsar 的干货和动态，请关注公众号 ApachePulsar。
点击“阅读原文”，查看原作者 blog。